package io;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hadoop.mapred.SequenceFileRecordReader;
import org.apache.hadoop.util.ReflectionUtils;

@SuppressWarnings("unchecked")
public class LineRecordReaderForSequenceFile<T> implements RecordReader<LongWritable, Text> {
	private SequenceFileRecordReader<T, Text> reader = null;
	private T tmp = null;	
	public LineRecordReaderForSequenceFile(Configuration conf, FileSplit split) throws IOException {
		reader = new SequenceFileRecordReader<T, Text>(conf, split);
		tmp = (T) ReflectionUtils.newInstance(reader.getKeyClass(), conf);
	}

	public boolean next(LongWritable key, Text value) throws IOException {
		key.set(reader.getPos());		
		return reader.next(tmp, value);
	}
  
	public LongWritable createKey() {
		return new LongWritable();
	}
	  
	public Text createValue() {
		return new Text();
	}

	public float getProgress() throws IOException {
		return reader.getProgress();
	}
	  
	public synchronized long getPos() throws IOException {
		return reader.getPos();
	}	  
	  
	public synchronized void close() throws IOException { reader.close(); }
}
